{#
  Template for 'request_codec_request_test.cc'.

  Provides integration tests using Kafka codec.
  The tests do the following:
  - create the message,
  - serialize the message into buffer,
  - pass the buffer to the codec,
  - capture messages received in callback,
  - verify that captured messages are identical to the ones sent.
#}
#include "extensions/filters/network/kafka/external/requests.h"
#include "extensions/filters/network/kafka/request_codec.h"

#include "test/extensions/filters/network/kafka/serialization_utilities.h"
#include "test/mocks/server/mocks.h"

#include "gtest/gtest.h"

namespace Envoy {
namespace Extensions {
namespace NetworkFilters {
namespace Kafka {
namespace RequestCodecRequestTest {

class RequestCodecRequestTest : public testing::Test {
protected:
  template <typename T> void putInBuffer(T arg);

  Buffer::OwnedImpl buffer_;
};

{% for request_type in request_types %}

// Integration test for {{ request_type.name }} messages.

TEST_F(RequestCodecRequestTest, shouldHandle{{ request_type.name }}Messages) {
  // given
  using RequestUnderTest = Request<{{ request_type.name }}>;

  std::vector<RequestUnderTest> sent;
  int32_t correlation = 0;

  {% for field_list in request_type.compute_field_lists() %}
  for (int i = 0; i < 100; ++i ) {
    const RequestHeader header =
      { {{ request_type.get_extra('api_key') }}, {{ field_list.version }}, correlation++, "id" };
    const {{ request_type.name }} data = { {{ field_list.example_value() }} };
    const RequestUnderTest request = {header, data};
    putInBuffer(request);
    sent.push_back(request);
  }
  {% endfor %}

  const InitialParserFactory& initial_parser_factory = InitialParserFactory::getDefaultInstance();
  const RequestParserResolver& request_parser_resolver =
    RequestParserResolver::getDefaultInstance();
  const CapturingRequestCallbackSharedPtr request_callback =
    std::make_shared<CapturingRequestCallback>();

  RequestDecoder testee{initial_parser_factory, request_parser_resolver, {request_callback}};

  // when
  testee.onData(buffer_);

  // then
  const std::vector<AbstractRequestSharedPtr>& received = request_callback->getCaptured();
  ASSERT_EQ(received.size(), sent.size());

  for (size_t i = 0; i < received.size(); ++i) {
    const std::shared_ptr<RequestUnderTest> request =
      std::dynamic_pointer_cast<RequestUnderTest>(received[i]);
    ASSERT_NE(request, nullptr);
    ASSERT_EQ(*request, sent[i]);
  }
}
{% endfor %}

template <typename T>
void RequestCodecRequestTest::putInBuffer(const T arg) {
  RequestEncoder encoder{buffer_};
  encoder.encode(arg);
}

} // namespace RequestCodecRequestTest
} // namespace Kafka
} // namespace NetworkFilters
} // namespace Extensions
} // namespace Envoy
